package queue

import (
	"fmt"
	"log"
	"server/internal/config"

	amqp "github.com/rabbitmq/amqp091-go"
	"github.com/zeromicro/go-queue/rabbitmq"
	"github.com/zeromicro/go-zero/core/logx"
	"github.com/zeromicro/go-zero/core/queue"
)

type (
	ConsumeHandle func(message string) error

	ConsumeHandler interface {
		Consume(message string) error
	}

	RabbitListener struct {
		conn    *amqp.Connection
		channel *amqp.Channel
		forever chan bool
		handler ConsumeHandler
		queues  rabbitmq.RabbitListenerConf
		config  config.Config
	}
)

func getRabbitURL(rabbitConf rabbitmq.RabbitConf) string {
	return fmt.Sprintf("amqp://%s:%s@%s:%d/%s", rabbitConf.Username, rabbitConf.Password,
		rabbitConf.Host, rabbitConf.Port, rabbitConf.VHost)
}

func MustNewListener(listenerConf rabbitmq.RabbitListenerConf, handler rabbitmq.ConsumeHandler, config config.Config) queue.MessageQueue {
	listener := RabbitListener{queues: listenerConf, handler: handler, forever: make(chan bool), config: config}
	url := getRabbitURL(listenerConf.RabbitConf)
	// logx.Infof("Starting rabbitmq listener to %s...", url)
	conn, err := amqp.Dial(url)
	if err != nil {
		log.Fatalf("failed to connect rabbitmq, error: %v", err)
	}

	listener.conn = conn
	channel, err := listener.conn.Channel()
	if err != nil {
		log.Fatalf("failed to open a channel: %v", err)
	}

	listener.channel = channel
	return listener
}

func (q RabbitListener) CreateQueue(queueName string) error {
	// 1. 首先声明	交换器
	err := q.channel.ExchangeDeclare(
		q.config.RabbitConf.Exchange, // 交换器名称
		"direct",                     // 交换器类型
		true,                         // 持久化
		false,                        // 自动删除
		false,                        // 内部(不供客户端使用)
		false,                        // 不等待
		nil,                          // 参数
	)
	if err != nil {
		return fmt.Errorf("Exchange Declare failed: %v", err)
	}

	_, err = q.channel.QueueDeclare(
		queueName,
		true,
		false,
		false,
		false,
		nil,
	)
	// logx.Infof("Queue Declare: %#v", res)
	if err != nil {
		return fmt.Errorf("Queue Declare: %s", err)
	}

	if err = q.channel.QueueBind(
		queueName,                    // name of the queue
		queueName,                    // bindingKey
		q.config.RabbitConf.Exchange, // sourceExchange
		false,                        // noWait
		nil,                          // arguments
	); err != nil {
		return fmt.Errorf("Queue Bind: %s", err)
	}
	return err
}

func (q RabbitListener) Start() {
	for _, que := range q.queues.ListenerQueues {
		err := q.CreateQueue(que.Name)
		if err != nil {
			logx.Errorf("failed to CreateQueue a queue, error: %v", err)
		}
		msg, err := q.channel.Consume(
			que.Name,
			"",
			que.AutoAck,
			que.Exclusive,
			que.NoLocal,
			que.NoWait,
			nil,
		)
		if err != nil {
			logx.Errorf("failed to listener, error: %v", err)
		}

		go func() {
			for d := range msg {
				if err = q.handler.Consume(string(d.Body)); err != nil {
					logx.Errorf("Error on consuming: %s, error: %v", string(d.Body), err)
				}
				if !que.AutoAck {
					err = d.Ack(false)
					if err != nil {
						logx.Errorf("Error on ack: %s, error: %v", string(d.Body), err)
					}
				}
			}
		}()
	}

	<-q.forever
}

func (q RabbitListener) Stop() {
	q.channel.Close()
	q.conn.Close()
	close(q.forever)
}
